package com.staryea.newcustomer.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collector;
import java.util.stream.Stream;

/**
 * @Auther: hs
 * @Date: 2019/3/17 21:57
 * @Description:
 */
@Slf4j
@Component
public class KafkaListeners {

    @KafkaListener(id = "${kafka.listener.group-id}",
            topics = {"${kafka.listener.topics}"},
            containerFactory = "kafkaListenerContainerFactory")
    public void batchListener(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        try {
            records.stream().limit(2).forEach(record -> {
                log.info("value: {}", record.value());
            });

        } catch (Exception e) {
            log.error("kafka监听异常:" + e.getMessage(), e);
        } finally {
            ack.acknowledge();
        }
    }
}
